package com.shadowlight.hero_story.mq;

import com.alibaba.fastjson.JSONObject;
import com.shadowlight.hero_story.config.AllConfig;
import com.shadowlight.hero_story.rank.RankService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

/**
 * @Description 消息队列消费者
 * @Author cz
 * @Date 2022/4/18
 */
@Slf4j
public final class MQConsumer {

    private MQConsumer() {
    }

    public static void init() {
        if (!AllConfig.ROCKETMQ_ENABLE) {
            log.debug("跳过消费者初始化!");
            return;
        }
        try {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("hero_story", getAclRPCHook(), new AllocateMessageQueueAveragely());
            consumer.subscribe("Victor", "*");
            consumer.setNamesrvAddr(AllConfig.ROCKETMQ_NAMESRV_ADDR);

            consumer.registerMessageListener((MessageListenerConcurrently) (msgList, ctx) -> {
                for (MessageExt ext : msgList) {
                    VictorMsg msg;
                    if (ext != null) {
                        msg = JSONObject.parseObject(ext.getBody(), VictorMsg.class);
                        log.info("mq收取到的信息,winnerId:{}, loserId:{}", msg.getWinnerId(), msg.getLoserId());
                        RankService.getInstance().refreshRank(msg.getWinnerId(), msg.getLoserId());
                    }
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
            log.info("mq连接成功");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }

    }


    //ACL相关key
    static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("heroStory", "heroStory6677@"));
    }
}
